package com.iceland.server.server.cmd;

import com.iceland.common.listener.ConsumeListener;
import com.iceland.common.listener.ConsumeQueue;
import com.iceland.common.thread.MessageEventExcutor;
import com.iceland.common.rpc.RpcMessageManager;
import com.iceland.common.rpc.RpcServiceInterfaceFactory;
import com.iceland.common.service.client.RemoteClientService;
import com.iceland.server.config.Environment;
import com.iceland.server.server.proxy.ProxyServer;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.MarkerFactory;

import java.util.concurrent.LinkedBlockingQueue;

@Slf4j
public class ChannelPool {
    public final int MAX_CHANNELS_LENGTH= Environment.getProxyConfig().getMaxProxyServerConnectionSize();

    //正在被使用的通道
    public volatile LinkedBlockingQueue<Channel> occupationChannels=new LinkedBlockingQueue<>();

    //空闲的通道
    private ConsumeQueue<Channel> availableChannels=new ConsumeQueue<Channel> (){
        @Override
        public boolean beforeConsume(Channel element, ConsumeListener listener) {
            if(element==null ){
                this.addListener(listener);
                return false;
            }
            if( !element.isActive()){
                element.close();
                this.addListener(listener);
                return false;
            }
            //将通道放入占据队列
            occupationChannels.add(element);
            return true;
        }
    };

    /**
     * 获取通道总和
     * @return
     */
    public int getChannelSize(){
        return availableChannels.size()+occupationChannels.size();
    }

    public int getOccupationChannelSize(){
        return occupationChannels.size();
    }
    public int getAvailableChannelSize(){
        return availableChannels.size();
    }

    public int getListnerSize(){
        return availableChannels.getListenerSize();
    }


    /**
     * 获取一个可用的通道，没有则发消息创建，此时会阻塞线程等待10s，但并不保证一定有可用的通道返回
     * @return null or Channel
     */
    public boolean getAvailableChannel(ProxyServer proxyServer, ConsumeListener<Channel> listener){
        if(availableChannels.size()+occupationChannels.size()+availableChannels.getListenerSize()<MAX_CHANNELS_LENGTH ){
            RpcMessageManager rpcMessageManager = proxyServer.getRpcMessageManager();
            RpcServiceInterfaceFactory rpcServiceInterfaceFactory = rpcMessageManager.getRpcServiceInterfaceFactory();
            RemoteClientService remoteClientService = rpcServiceInterfaceFactory.getBean(RemoteClientService.class);
            MessageEventExcutor.getInstance().submit(new Runnable() {
                @Override
                public void run() {
                    try{
                        log.debug(MarkerFactory.getMarker(ChannelPool.this.getClass().getTypeName()),"代理服务器{}发送建立一条流量转发通道",proxyServer.getPort());
                        remoteClientService.createDataExchangeChannel(proxyServer.getPort());
                    }
                    catch (Throwable e){
                        log.error("{}",e);
                    }
                }
            });
            availableChannels.addListener(listener);
            return true;
        }
        return false;

    }

    /**
     * 添加空闲通道
     */
    public void addAvailableChannel(Channel channel){
        availableChannels.add(channel);
    }


    /**
     * 释放被占用的通道，通道直接关闭
     * @param channel
     */
    public void freeOccupationChannel(Channel channel){
        if(occupationChannels.remove(channel)){
//            if(channel.isActive()) {
//                addAvailableChannel(channel);
//            }
//            else {
                channel.close();
//            }
        }
    }
//
//    /**
//     * 关闭并移除通道,注意：要释放该通道请先占用通道
//     * @param channel
//     * @return 当通道刚被释放就立即被占用会导致移除失败
//     */
//    public boolean closeChannel(Channel channel){
//        if(occupationChannels.remove(channel)){
//            channel.close();
//            return true;
//        }
//        return false;
//    }



}
